package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 * 服务实现类
 * 优惠券订单相关
 * </p>
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder>
        implements IVoucherOrderService {

    // lua脚本
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    @Resource
    private ISeckillVoucherService seckillVoucherService;
    @Resource
    private RedisIdWorker redisIdWorker;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
//    @Resource
//    private RedissonClient redissonClient;

    // 异步处理需要的线程池
    // 创建一个单线程化的线程池，即线程池中只有一个工作线程
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    // 定义了一个线程安全的阻塞队列 orderTasks，用于存储 VoucherOrder 类型的对象
//    private final BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
//    private IVoucherOrderService proxy;

    // 被 @PostConstruct 标注的方法，会在对象实例化完成后（构造方法执行后）、依赖注入（DI）完成后自动执行，且仅执行一次。
    @PostConstruct
    private void init() {
        // 启动前确保消费者组存在（Stream流 已由 Lua 创建），避免 NOGROUP 异常
        try {
            String streamKey = "stream.orders";
            String group = "g1";
            try {
                // 从 0 位点创建消费者组，处理历史未确认消息
                stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), group);
            } catch (Exception e) {
                // BUSYGROUP 等已存在情形，忽略
            }
        } catch (Exception e) {
            log.error("初始化 Stream/Group 失败: {}", e.getMessage());
        }

        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    // 用于线程池处理的任务：初始化完成之后，就回去队列中拿取信息
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // 1.获取队列中的订单信息
//                    VoucherOrder voucherOrder = orderTasks.take();
//1.获取消息队列中的订单信息:XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1
//                    以消费者组 g1 中的消费者 c1 身份读取
//                    每次最多读取 1 条消息
//                    如果没有消息，会阻塞等待 2 秒（超时后返回空）
//                    从上次消费的位置继续读取（不会重复消费已确认的消息）
//                    示例可传参数：
//                    List<MapRecord<K, HK, HV>> read(
//                            Consumer consumer,                // 1. 消费者信息（消费者组 + 消费者名称）
//                            StreamReadOptions readOptions,    // 2. 读取选项（如读取数量、阻塞时间）
//                            StreamOffset<K>... streamOffsets  // 3. 流偏移量（指定要读取的 Stream 及起始位置）
//);
//                    用于从名为 stream.orders 的流中读取消息，属于消费者组 g1 中的消费者 c1。
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
//                            如果消费者组 g1 不存在，直接执行这段代码不会会自动创建消费者组 g1，而是会抛出异常
//                            但是消费者c1不需要显示创建，首次使用会被自动注册
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
//                    2.判断订单信息是否为空
                    if (list == null || list.isEmpty()) {
//                        如果为null，说明没有消息，继续执行下一次循环
                        continue;
                    }
//                    3.非空，就是有消息，解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
//                    将 Map 中的键值对数据填充到 Java Bean 对象中
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 4、创建订单
                    createVoucherOrder(voucherOrder);
//                    5.发送确定XACK
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
                } catch (Exception e) {
                    log.error("订单处理时的异常：{}", e.getMessage());
                    String msg = e.getMessage();
                    if (msg != null && (msg.contains("Redisson is shutdown") || msg.contains("Event executor group is terminated"))) {
                        break;
                    }
                    //处理异常消息
                    handlePendingList();
                }
            }
        }
    }

    private void handlePendingList() {
        while (true) {
            try {
//                1.获取pending-list中的订单信息
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                );
//                2,判断消息是否为空
                if (list == null || list.isEmpty()) {
                    break;//没有消息，就结束循环
                }
                //解析数据
                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                createVoucherOrder(voucherOrder);
//                既可以对单个确认，又可以多个消息确认
//                Long acknowledge(
//    String streamKey,       // 1. Stream 的键名（消息所在的队列）
//    String consumerGroup,   // 2. 消费者组名称
//    RecordId recordId       // 3. 要确认的消息 ID
//);
                stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
            } catch (Exception e) {
                log.error("处理pendding订单异常", e);
                String msg = e.getMessage();
                if (msg != null && (msg.contains("Redisson is shutdown") || msg.contains("Event executor group is terminated"))) {
                    break;
                }
                try {
                    Thread.sleep(20);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    @javax.annotation.PreDestroy
    private void onDestroy() {
        try {
            SECKILL_ORDER_EXECUTOR.shutdownNow();
        } catch (Exception ignore) {
        }
    }
  
//    private void handleVoucherOrder(VoucherOrder voucherOrder) throws InterruptedException {
//        Long userId = voucherOrder.getUserId();
//        RLock redissionLock = redissonClient.getLock("lock:order:" + userId);
//        boolean isLock = redissionLock.tryLock(0, 5, TimeUnit.SECONDS);
//        if (!isLock) {
//            log.error("获取锁失败,不允许重复下单");
//            return;
//        }
//        try {
//            // 通过代理对象调用，事务生效
//            proxy.createVoucherOrder(voucherOrder);
//        } finally {
//            redissionLock.unlock();
//        }
//    }

    // 优化之后的写法
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");

        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(),
                userId.toString(), String.valueOf(orderId));

        String r = result != null ? result.toString() : "0";
        int resultInt = Integer.parseInt(r);

        if (resultInt != 0) {
            // 没有资格秒杀资格
            return Result.fail("下单失败：" + (resultInt == 1 ? "无库存" : "重复下单"));
        }
        // 有下单资格，执行下单逻辑:新开线程来执行下单任务，不断从阻塞队伍中获取信息（优惠券id，用户id），异步下单
        // 思考:下单的话需要什么呢,需要用户id和优惠券id,需要进行数据库操作,在这里需要开启一个异步线程,需要从消息队列中获取消息
        // 1.封装一个订单对象

//        VoucherOrder voucherOrder = new VoucherOrder();
//        voucherOrder.setId(orderId);
//        voucherOrder.setVoucherId(voucherId);
//        voucherOrder.setUserId(userId);
//        // 获取当前代理对象
//        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 放入阻塞队列中
//        orderTasks.add(voucherOrder);
        // 3.获取代理对象
        return Result.ok(orderId);

    }

    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        Long voucherId = voucherOrder.getVoucherId();
        Long userId = voucherOrder.getUserId();
        // 判断当前用户是否已经买过这个秒杀优惠券了
        int count = query().eq("voucher_id", voucherId).eq("user_id",
                userId).count();
        if (count > 0) {
            log.error("您已经购买过一次了,不能重复购买");
            return;
        }

        // 4.扣减库存
        // 都有效且当前用户没买过就减去一个库存
        /*
         * eq() 表示 等于 条件，即 WHERE voucher_id = #{voucherId}。
         * ge() 表示 大于等于 条件，即 WHERE stock>=0。
         * gt() 表示 大于 条件，即 WHERE stock>0。
         */
        // 乐观锁的思路
        boolean success = seckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherId)
                .gt("stock", 0).update();
        if (!success) {
            log.error("秒杀失败,库存不足");
            return;
        }
        // 4.创建一个优惠券订单
        save(voucherOrder);

    }

    // @Override
    // public Result seckillVoucher(Long voucherId) throws InterruptedException {
    // SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
    // // 1.是否在有效时间期
    //// if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
    //// return Result.fail("秒杀券活动还没开始！");
    //// }
    //// if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
    //// return Result.fail("秒杀券活动已经结束了！");
    //// }
    //// // 2.是否还有库存
    //// if (seckillVoucher.getStock() < 1) {
    //// return Result.fail("库存不足，秒杀券已被抢光！");
    //// }
    // // 3.一人一单
    // UserDTO user = UserHolder.getUser();
    // Long userId = user.getId();
    // // 创建锁对象
    //// SimpleRedisLock lock = new SimpleRedisLock("order:" + userId,
    // stringRedisTemplate);
    // RLock lock = redissonClient.getLock("lock:order:" + userId);
    //
    //// log.info("线程{}准备抢锁", Thread.currentThread().getId());
    // boolean isLock = lock.tryLock(0, 5, TimeUnit.SECONDS);
    //// log.info("线程{}获取锁{}", Thread.currentThread().getId(), isLock ? "成功" :
    // "失败");
    //
    // if (!isLock) {
    // // 采用非阻塞式锁，获取失败了就直接返回错误消息
    // return Result.fail("不允许重复下单");
    // }
    // try {
    // return createVoucherOrder(voucherId, userId);
    // } finally {
    // // 释放锁
    // lock.unlock();
    // }
    //
    // }

}
